package com.atguigu.flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/22 0022 21:19
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 下游从上游收到了多个Barrier，何时进行ck?
 * 设置Barrier对齐
 * 开始对齐后，当前Barrier后的数据，不处理，缓存起来，EOS，保证准确一次，实时性可能打折扣。
 * 开始对齐后，当前Barrier后的数据，继续处理，at least once。 保证实时性。
 * <p>
 * 默认 aligned Checkpoints，精确一次。
 * <p>
 * <p>
 * 设置Barrier不对齐
 * 只支持精确一次
 */
public class Flink14_CheckPointConfig {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        conf.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, false);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        //设置ck为本地文件系统
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck");

        //两次ck之间，最少暂停30s
        checkpointConfig.setMinPauseBetweenCheckpoints(30000);

        //设置ck的超时时间为1分钟，指一次备份的全流程的时间
        checkpointConfig.setCheckpointTimeout(60000);

        //如果是不对齐的ck，这个参数必须为1
        checkpointConfig.setMaxConcurrentCheckpoints(1);

        //只有开启了允许非对齐，这个参数才有意义，setAlignedCheckpointTimeout(0),代表使用非对齐进行ck
        // setAlignedCheckpointTimeout(x > 0),代表一开始使用对齐的方式ck，但是如果对齐的方式ck时间超过x，切换到非对齐的方式。
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(10));

        //默认清空，job被cacel后，ck的数据是删除的。 设置保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //ck连续失败3次，Job就停止
        checkpointConfig.setTolerableCheckpointFailureNumber(3);

        //开启不对齐
        checkpointConfig.enableUnalignedCheckpoints();

        //开启了checkpoint(备份)，每间隔2s自动备份。程序就会无限次重启
        env.enableCheckpointing(2000);

        env
                .socketTextStream("127.0.0.1", 8888)
                .map(new MyMapFunction())
                .addSink(new SinkFunction<String>() {
                    @Override
                    public void invoke(String value, Context context) throws Exception {
                        if (value.contains("x")) {
                            //人为手动抛出一个异常，为了模拟故障的情况
                            throw new RuntimeException("出异常了!");
                        }
                        System.out.println(value);
                    }
                });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    public static class MyMapFunction implements MapFunction<String, String>, CheckpointedFunction {

        ListState<String> strs;

        @Override
        public String map(String s) throws Exception {
            //存
            strs.add(s);
            //取
            return strs.get().toString();
        }

        @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();
            ListStateDescriptor<String> listStateDescriptor1 = new ListStateDescriptor<>("list1", String.class);

            strs = operatorStateStore.getListState(listStateDescriptor1);
        }
    }

}
